import com.alibaba.fastjson.JSONObject;
import com.auxgroup.smarthome.mqtt.AuxMqttClient;
import com.auxgroup.smarthome.mqtt.MqttConfProp;
import org.fusesource.mqtt.client.MQTTException;
import org.fusesource.mqtt.client.QoS;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by kevinchen on 2017/8/11.
 * 模拟几个安卓客户端发布 app用户距离信息
 */
public class MQTTAndroidSendDistanceInfo {
    public final static Map<String, AtomicInteger> uidMap = new ConcurrentHashMap<>();
    private static final Logger log = LoggerFactory.getLogger(MQTTAndroidSendDistanceInfo.class);
    private static AuxMqttClient auxMqttClient = null;

    static {
        uidMap.put("1000a1", new AtomicInteger());
        uidMap.put("1000a2", new AtomicInteger());
        uidMap.put("1000a3", new AtomicInteger());
        uidMap.put("1000a4", new AtomicInteger());
        uidMap.put("1000a5", new AtomicInteger());
    }

    public static void main(String[] args) {
        try {
            MqttConfProp mqttConfProp = new MqttConfProp();
            mqttConfProp.setHost("127.0.0.1");
            mqttConfProp.setPort(1886);
            auxMqttClient = new AuxMqttClient(mqttConfProp);
            auxMqttClient.connect();
            ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(4);
            for (Iterator<Map.Entry<String, AtomicInteger>> iterator = uidMap.entrySet().iterator(); iterator.hasNext(); ) {
                Map.Entry<String, AtomicInteger> next = iterator.next();
                final String uid = next.getKey();
                final AtomicInteger distance = next.getValue();
                PublishTask publishTask = new PublishTask(auxMqttClient, uid, distance);
                scheduledExecutorService.scheduleAtFixedRate(publishTask, 0, 10, TimeUnit.SECONDS);
            }
        } catch (Exception e) {
            log.debug("Mqtt exception", e);
        }
    }


    /**
     * 安卓客户端发布消息
     */
    static class PublishTask implements Runnable {

        private final AuxMqttClient auxMqttClient;

        private final String uid;

        private final AtomicInteger distance;

        public PublishTask(final AuxMqttClient auxMqttClient, final String uid, final AtomicInteger distance) {
            this.uid = uid;
            this.distance = distance;
            this.auxMqttClient = auxMqttClient;
        }

        @Override
        public void run() {

            String topic = "mqtt/distance/";
            //主题的内容
            JSONObject jsonObject = new JSONObject();
            jsonObject.put("uid", uid);
            jsonObject.put("distance", distance.incrementAndGet());
            String jsonMsg = jsonObject.toJSONString();
            try {
                auxMqttClient.publish(topic, jsonMsg.getBytes(), QoS.AT_MOST_ONCE, false);
            } catch (MQTTException e) {
                log.debug("MQTT publish exception", e);
            }
        }
    }

}
